package org.wj.accumulator

import org.apache.spark.util.AccumulatorV2

class UserAccumulator[User] extends AccumulatorV2[User, Array[Int]] {

  private val array: Array[Int] = Array(0, 0)

  override def isZero: Boolean = array.mkString.toLong == 0L

  override def copy(): AccumulatorV2[User, Array[Int]] = {
    val copyToLocal = new UserAccumulator[User]()
    array.copyToArray(copyToLocal.array)
    copyToLocal
  }

  override def reset(): Unit = {
    for (i <- array.indices) {
      array(i) = 0
    }
  }

  override def add(v: User): Unit = {
    v match {
      case User(_, "男", _) => array(0) += 1
      case User(_, "女", _) => array(1) += 1
    }

  }

  override def merge(other: AccumulatorV2[User, Array[Int]]): Unit = {
    val o = other.asInstanceOf[UserAccumulator[User]]
    array(0) += o.array(0)
    array(1) += o.array(1)
  }

  override def value: Array[Int] = array

  override def toString(): String = array.toList.toString()

}

object UserAccumulator extends App {


  private val _array: Array[Int] = Array(0, 0, 0, 0)

  private val a_ = 1


  for (i <- _array) {
    println(i)
  }

}